package com.pw.study.flink.sink;

import com.alibaba.fastjson.JSON;
import com.pw.study.flink.entities.WaterSensor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.redis.RedisSink;
import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisPoolConfig;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommand;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommandDescription;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisMapper;

public class MyRedisSink {
    public static void main(String[] args) {
        //dealRedis();
        //dealRedis2();
        dealRedis3();
    }
    private static void dealRedis3() {
        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 20000);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        env.setParallelism(2);
        DataStreamSource<WaterSensor> stream = env.fromElements(
                new WaterSensor("s1", 1L, 10),
                new WaterSensor("s1", 3L, 30),
                new WaterSensor("s1", 4L, 30),
                new WaterSensor("s1", 2L, 20),
                new WaterSensor("s_1", 5L, 40),
                new WaterSensor("s_2", 4L, 100),
                new WaterSensor("se", 5L, 200)

        );

        FlinkJedisPoolConfig jedisConf = new FlinkJedisPoolConfig.Builder()
                .setHost("hadoop162")
                .setPort(6379)
                .setMaxTotal(100)
                .setMaxIdle(10)
                .setTimeout(10 * 1000)
                .setDatabase(0)
                .build();
        stream.keyBy(WaterSensor::getId).sum("vc").print();
        stream.addSink(new RedisSink<>(jedisConf, new RedisMapper<WaterSensor>() {
            @Override
            public RedisCommandDescription getCommandDescription() {
                // 第二个参数只对hash和zset有效, 其他数据结构忽略
                System.out.println("add................");
                return new RedisCommandDescription(RedisCommand.HSET, "hello");
            }

            @Override
            public String getKeyFromData(WaterSensor waterSensor) {
                System.out.println("key...................");
                return waterSensor.getId();
            }

            @Override
            public String getValueFromData(WaterSensor waterSensor) {
                System.out.println("value............................");
                return JSON.toJSONString(waterSensor);
            }
        }));

        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
    private static void dealRedis2() {
        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 20000);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        DataStreamSource<WaterSensor> stream = env.fromElements(
                new WaterSensor("sensor_1", 1L, 10),
                new WaterSensor("sensor_1", 3L, 30),
                new WaterSensor("sensor_1", 4L, 30),
                new WaterSensor("sensor_1", 2L, 20),
                new WaterSensor("sensor_1", 5L, 40),
                new WaterSensor("sensor_2", 4L, 100),
                new WaterSensor("sensor_2", 5L, 200)

        );

        FlinkJedisPoolConfig jedisConf = new FlinkJedisPoolConfig.Builder()
                .setHost("hadoop162")
                .setPort(6379)
                .setMaxTotal(100)
                .setMaxIdle(10)
                .setTimeout(10 * 1000)
                .setDatabase(0)
                .build();
        stream.keyBy(WaterSensor::getId).sum("vc").print();
        stream.addSink(new RedisSink<>(jedisConf, new RedisMapper<WaterSensor>() {
            @Override
            public RedisCommandDescription getCommandDescription() {
                System.out.println("add................");
                return new RedisCommandDescription(RedisCommand.SADD, null);
            }

            @Override
            public String getKeyFromData(WaterSensor waterSensor) {
                System.out.println("key...................");
                return waterSensor.getId();
            }

            @Override
            public String getValueFromData(WaterSensor waterSensor) {
                System.out.println("value............................");
                return JSON.toJSONString(waterSensor);
            }
        }));

        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    private static void dealRedis() {
        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 20000);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        DataStreamSource<WaterSensor> stream = env.fromElements(
                new WaterSensor("s1_1", 1L, 10),
                new WaterSensor("s1_1", 3L, 30),
                new WaterSensor("s1_1", 4L, 30),
                new WaterSensor("se1_1", 2L, 20),
                new WaterSensor("se1_1", 5L, 40),
                new WaterSensor("s1_2", 4L, 100),
                new WaterSensor("s1_2", 5L, 200),
                new WaterSensor("hello", 5L, 200)
        );

        FlinkJedisPoolConfig jedisConf = new FlinkJedisPoolConfig.Builder()
                .setHost("hadoop162")
                .setPort(6379)
                .setMaxTotal(100)
                .setMaxIdle(10)
                .setTimeout(10 * 1000)
                .setDatabase(0)
                .build();

        stream.addSink(new RedisSink<>(jedisConf, new RedisMapper<WaterSensor>() {

            @Override
            public RedisCommandDescription getCommandDescription() {
                // 第二个参数只对hash和zset有效, 其他数据结构忽略
                return new RedisCommandDescription(RedisCommand.SET, null);
            }

            // 数据要写入到redis.
            @Override
            public String getKeyFromData(WaterSensor data) {
                return data.getId();  // sensor id作为字符串的key
            }

            @Override
            public String getValueFromData(WaterSensor data) {
                return JSON.toJSONString(data);
            }
        }));


        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
